use std::io::Read;
use std::sync::Arc;
use std::{io, thread};

use futures::channel::oneshot;
use futures::executor::block_on;
use futures::prelude::*;

use grpcio::{ChannelBuilder, Environment, ResourceQuota, RpcContext, ServerBuilder, UnarySink};

use grpcio_proto::example::helloworld::{HelloReply, HelloRequest};

use super::protos::request::{ProposalRequest, ProposalResponse, APIRequest, APIResponse, MessageResponse};
use crate::protos::eraftpb::Message;
use std::sync::mpsc::TryRecvError;

/// 要实现
/// 1. client 给 server发请求  req: APIRequest, resp: APIResponse
/// 2. server 给 server发请求  req: Message, resp: MessageResponse
pub trait NodeService {
    fn do_(&mut self, ctx: ::grpcio::RpcContext, req: APIRequest, sink: ::grpcio::UnarySink<APIResponse>);

    fn inner_do(&mut self, ctx: ::grpcio::RpcContext, req: Message, sink: ::grpcio::UnarySink<MessageResponse>);
}


const METHOD_NODE_DO_: ::grpcio::Method<APIRequest, APIResponse> = ::grpcio::Method {
    ty: ::grpcio::MethodType::Unary,
    name: "/kv/do",
    req_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
    resp_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
};


const METHOD_NODE_INNER_DO_: ::grpcio::Method<Message, MessageResponse> = ::grpcio::Method {
    ty: ::grpcio::MethodType::Unary,
    name: "/kv/inner-do",
    req_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
    resp_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
};


#[derive(Clone)]
pub struct NodeServiceV1;


impl NodeService for NodeServiceV1 {
    fn do_(&mut self, ctx: RpcContext<'_>, req: APIRequest, sink: UnarySink<APIResponse>) {
        unimplemented!()
    }

    fn inner_do(&mut self, ctx: RpcContext<'_>, req: Message, sink: UnarySink<MessageResponse>) {
        match req {
            Ok(msg) => node.step(msg, &logger),
            Err(TryRecvError::Empty) => return,
            Err(TryRecvError::Disconnected) => return,
        }
        let mut resp = MessageResponse::default();
        resq.success = true;
        let f = sink
            .success(resp)
            .map_err(move |e| error!("failed to reply {:?}: {:?}", req, e))
            .map(|_| ());
        ctx.spawn(f)
    }
}


pub fn create_node_service<S: NodeService + Send + Clone + 'static>(s: S) -> ::grpcio::Service {
    let mut builder = ::grpcio::ServiceBuilder::new();
    let mut instance = s;
    builder = builder.add_unary_handler(&METHOD_NODE_DO_, move |ctx, req, resp| {
        instance.do_(ctx, req, resp)
    });
    builder = builder.add_unary_handler(&METHOD_NODE_INNER_DO_, move |ctx, req, resp| {
        instance.inner_do(ctx, req, resp)
    });
    builder.build()
}


const METHOD_GREETER_SAY_HELLO: ::grpcio::Method<super::helloworld::HelloRequest, super::helloworld::HelloReply> = ::grpcio::Method {
    ty: ::grpcio::MethodType::Unary,
    name: "/helloworld.Greeter/SayHello",
    req_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
    resp_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
};


pub trait Greeter {
    fn say_hello(&mut self, ctx: ::grpcio::RpcContext, req: super::helloworld::HelloRequest, sink: ::grpcio::UnarySink<super::helloworld::HelloReply>);
}

pub fn create_greeter<S: Greeter + Send + Clone + 'static>(s: S) -> ::grpcio::Service {
    let mut builder = ::grpcio::ServiceBuilder::new();
    let mut instance = s;
    builder = builder.add_unary_handler(&METHOD_GREETER_SAY_HELLO, move |ctx, req, resp| {
        instance.say_hello(ctx, req, resp)
    });
    builder.build()
}

#[derive(Clone)]
struct GreeterService;

impl Greeter for GreeterService {
    fn say_hello(&mut self, ctx: RpcContext<'_>, req: HelloRequest, sink: UnarySink<HelloReply>) {
        let msg = format!("Hello {}", req.get_name());
        let mut resp = HelloReply::default();
        resp.set_message(msg);
        let f = sink
            .success(resp)
            .map_err(move |e| error!("failed to reply {:?}: {:?}", req, e))
            .map(|_| ());
        ctx.spawn(f)
    }
}

fn main() {
    let _guard = log_util::init_log(None);
    let env = Arc::new(Environment::new(1));
    let service = create_greeter(GreeterService);

    let quota = ResourceQuota::new(Some("HelloServerQuota")).resize_memory(1024 * 1024);
    let ch_builder = ChannelBuilder::new(env.clone()).set_resource_quota(quota);

    let mut server = ServerBuilder::new(env)
        .register_service(service)
        .bind("127.0.0.1", 50_051)
        .channel_args(ch_builder.build_args())
        .build()
        .unwrap();
    server.start();
    for (host, port) in server.bind_addrs() {
        info!("listening on {}:{}", host, port);
    }
    let (tx, rx) = oneshot::channel();
    thread::spawn(move || {
        info!("Press ENTER to exit...");
        let _ = io::stdin().read(&mut [0]).unwrap();
        tx.send(())
    });
    let _ = block_on(rx);
    let _ = block_on(server.shutdown());
}